package day10

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import java.io.{BufferedReader, InputStream, InputStreamReader}
import java.net.{ConnectException, Socket}
import java.nio.charset.StandardCharsets
import scala.collection.mutable

/**
 * 第3章	DStream创建
 *
 * 3.1	RDD队列(了解)
 *
 * 3.1.1	用法及说明
 *
 * 测试过程中，可以通过使用ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到这个队列中的RDD，都会作为一个DStream处理。
 *
 */
object Spark_Streaming_3 {
}

/**
 * 3.1.2	案例实操
 *
 * 1)	需求：循环创建几个RDD，将RDD放入队列。通过SparkStream创建DStream，计算WordCount
 * 2)	编写代码
 */
object Spark_Streaming_3_1 {
  def main(args: Array[String]): Unit = {
    // 创建Spark配置信息对象
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]")
    // 创建SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 创建RDD队列
    val rddQueue: mutable.Queue[RDD[Int]] = new mutable.Queue[RDD[Int]]
    // 创建QueueInputDStream
    val inputStream: InputDStream[Int] = ssc.queueStream(queue = rddQueue, oneAtATime = false)
    // 转换数据的结构
    val mapDStream: DStream[(Int, Int)] = inputStream.map((_, 1))
    // 调用reduceByKey
    val reduceDStream: DStream[(Int, Int)] = mapDStream.reduceByKey(_ + _)
    // 打印结果
    reduceDStream.print()
    // 启动任务
    ssc.start()
    // 循环创建并向RDD队列中放入RDD
    for (i <- 1 to 10) {
      val sc: SparkContext = ssc.sparkContext
      rddQueue += sc.makeRDD(seq = 1 to 5, numSlices = 10)
      Thread.sleep(2000)
    }
    // 等待采集结束，终止上下文环境对象
    ssc.awaitTermination()
  }
}

/**
 * 3.2	自定义数据源
 *
 * 3.2.1	用法及说明
 * 需要继承Receiver，并实现onStart、onStop方法来自定义数据源采集。
 *
 * 3.2.2	案例实操
 * 1)	需求：自定义数据源，实现监控某个端口号，获取该端口号内容。
 *
 * 2)	自定义数据源
 *
 * @param host 主机
 * @param port 端口号
 */
class MyReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {

  // 创建一个Socket
  private var socket: Socket = _

  // 最初启动的时候，调用该方法，作用为：读数据并将数据发送给Spark
  override def onStart(): Unit = {
    new Thread("Socket Receiver") {
      setDaemon(true)

      override def run(): Unit = {
        receive()
      }
    }.start()
  }

  // 读取数据并将数据发送给Spark
  def receive(): Unit = {
    try {
      socket = new Socket(host, port)
      // 创建一个BufferedReader用于读取端口传来的数据
      val inputStream: InputStream = socket.getInputStream
      val inputStreamReader: InputStreamReader = new InputStreamReader(inputStream, StandardCharsets.UTF_8)
      val reader: BufferedReader = new BufferedReader(inputStreamReader)
      // 定义一个变量，用来接受端口传过来的数据
      var input: String = null

      // 读取数据，循环发送数据给Spark。一般要想结束，发送特定的数据，如：“==END==”
      while ((input = reader.readLine()) != null) {
        store(input)
      }
    } catch {
      case e: ConnectException =>
        restart(s"Error connecting to $host:$port", e)
    }
  }

  override def onStop(): Unit = {
    if (socket != null) {
      socket.close()
      socket = null
    }
  }
}

/**
 * 使用自定义数据源采集数据
 */
object Spark_Streaming_3_2 {
  def main(args: Array[String]): Unit = {
    // 初始化Spark配置信息
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming_customDataSource").setMaster("local[*]")
    // 初始化SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 创建自定义receiver的Streaming
    val lineDStream: ReceiverInputDStream[String] = ssc.receiverStream(receiver = new MyReceiver(host = "hadoop102", port = 9999))
    // 将每一行数据做切分，形成一个个单词
    val wordDStream: DStream[String] = lineDStream.flatMap(_.split("\t"))
    // 将单词映射成元组
    val tupleDStream: DStream[(String, Int)] = wordDStream.map((_, 1))
    // 将相同的单词次数做统计
    val reduceDStream: DStream[(String, Int)] = tupleDStream.reduceByKey(_ + _)
    // 打印
    reduceDStream.print(num = 1000)
    // 启动SparkStreamingContext
    ssc.start()

    ssc.awaitTermination()
  }
}

/**
 * 3.3	Kafka数据源（面试开发重点）
 *
 * 3.3.1	版本选型
 *
 * ReceiverAPI：需要一个专门的Executor去接收数据，然后发送给其他的Executor做计算。
 * 存在的问题：接收数据的Executor和计算的Executor速度会有所不同，特别在接收数据的Executor速度大于计算的Executor速度，会导致计算数据的节点内存溢出。
 *
 * DirectAPI：是由计算的Executor来主动消费Kafka的数据，速度由自身控制。
 *
 */
object Spark_Streaming_3_3_1 {
}

/**
 * 3.3.2	Kafka 0-8 Receive模式
 *
 * 1）需求：通过SparkStreaming从Kafka读取数据，并将读取过来的数据做简单计算，最终打印到控制台。
 *
 * 0-8Receive模式，offset维护在zk中，程序停止后，继续生产数据，再次启动程序，仍然可以继续消费。
 * 可通过get /consumers/bigdata/offsets/主题名/分区号 查看
 */
object Spark_Streaming_3_3_2 {
  def main(args: Array[String]): Unit = {
    // 1、创建SparkConf
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_3_3_2").setMaster("local[*]")
    // 2、创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 3、使用ReceiverAPI读取Kafka数据创建DSteam
    // Zookeeper quorum机制
    val zkQuorum: String = "hadoop102:2181,hadoop103:2181,hadoop104:2181"
    // 消费者组id
    val groupId: String = "bigdata"
    // key：topic；value：topic对应的分区数
    val topics: Map[String, Int] = Map("my-bak" -> 2)
    val inputDStream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)
    // 4、计算wordCount并打印
    inputDStream.map(_._2)
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()
    // 5、开启任务
    ssc.start()
    ssc.awaitTermination()
  }
}

/**
 * 3.3.3	Kafka 0-8 Direct模式
 *
 * 1）需求：通过SparkStreaming从Kafka读取数据，并将读取过来的数据做简单计算，最终打印到控制台。
 *
 * 0-8 DirectAPI 自动维护offset1
 * offset维护在checkpoint中，但是获取StreamingContext的方式需要改变，目前这种方式会丢失数据
 */
object Spark_Streaming_3_3_3_1 {
  def main(args: Array[String]): Unit = {
    // 1、创建SparkConf对象
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_3_3_3_1").setMaster("local[*]")
    // 2、创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 3、准备kafka参数
    val kafkaParams: Map[String, String] = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata"
    )
    // 4、使用DirectAPI自动维护offset的方式获取kafka数据创建DStream
    val kafkaDStream: InputDStream[(String, String)] =
      KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics = Set("my-bak"))
    // 5、计算wordCount并打印
    kafkaDStream.map((_._2))
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()
    // 6、开启任务
    ssc.start()
    ssc.awaitTermination()
  }
}

/**
 * 0-8 DirectAPI 自动维护offset2
 *
 * offsset维护在checkpoint中，获取StreamingContext为getActiveOrCreate
 *
 * 这种方式缺点：
 * 	checkpoint小文件过多
 * 	checkpoint记录最后一次时间戳，再次启动的时候会把间隔时间的周期再执行一次
 */
object Spark_Streaming_3_3_3_2 {
  def main(args: Array[String]): Unit = {
    val checkPointPath: String = "D:\\dev\\workspace\\my-bak\\spark-bak\\cp"
    val ssc: StreamingContext = StreamingContext.getActiveOrCreate(checkPointPath, () => getStreamingContext)
    // 开启任务
    ssc.start()
    ssc.awaitTermination()
  }

  /**
   * 创建StreamingContext
   *
   * @return StreamingContext
   */
  def getStreamingContext: StreamingContext = {
    // 1、创建SparkConf对象
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_3_3_3_2").setMaster("local[*]")
    // 2、创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 3、设置检查点
    ssc.checkpoint("D:\\dev\\workspace\\my-bak\\spark-bak\\cp")
    // 4、准备kafka参数
    val kafkaParams: Map[String, String] = Map(
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata"
    )
    // 5、使用DirectAPI自动维护offset的方式读取kafka数据创建DStream
    val kafkaDStream: InputDStream[(String, String)] =
      KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics = Set("my-bak"))
    // 6、计算wordCount并打印
    kafkaDStream.map((_._2))
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()
    // 7、返回结果
    ssc
  }
}

/**
 * 0-8 DirectAPI  手动维护offset
 */
object Spark_Streaming_3_3_3_3 {
  def main(args: Array[String]): Unit = {
    // 1、创建SparkConf
    val conf: SparkConf = new SparkConf().setAppName("Spark_Streaming_3_3_3_3").setMaster("local[*]")
    // 2、创建SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 3、创建kafka参数
    val kafkaParams: Map[String, String] = Map[String, String](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9092,hadoop103:9092,hadoop104:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "bigdata"
    )
    // 4、获取上一次消费的位置信息(实际开发的场景中是从mysql数据库中读取offset)
    val fromOffsets: Map[TopicAndPartition, Long] = Map[TopicAndPartition, Long](
      TopicAndPartition("my-bak", 0) -> 13L,
      TopicAndPartition("my-bak", 1) -> 10L
    )
    // 5、使用DirectAPI手动维护offset的方式消费数据
    val kafkaDStream: InputDStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, String](
      ssc = ssc,
      kafkaParams = kafkaParams,
      fromOffsets = fromOffsets,
      messageHandler = (messageHandler: MessageAndMetadata[String, String]) => messageHandler.message())
    // 6、定义空集合用于存放数据的offset
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]
    // 7、将当前消费到的offset进行保存(实际开发的场景中是从将offset写入到mysql中)
    kafkaDStream.transform { rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
    }.foreachRDD {
      _ =>
        for (o <- offsetRanges) {
          println(s"${o.fromOffset}-${o.untilOffset}")
        }
    }

    // 8、开启任务
    ssc.start()
    ssc.awaitTermination()
  }
}

































